package tcp

import (
	"JoRedis/cluster"
	"JoRedis/config"
	"JoRedis/constant"
	"JoRedis/database"
	database2 "JoRedis/interface/database"
	atomic2 "JoRedis/lib/atomic"
	"JoRedis/lib/logger"
	"JoRedis/protocol"
	"JoRedis/redis"
	"bufio"
	"context"
	"fmt"
	"io"
	"net"
	"sync"
)

type RedisServer struct {
	// 所有的客户端连接
	activeConnMap sync.Map
	// redis服务是否正在关闭中
	closing atomic2.Boolean
	// 底层数据存储使用单机还是集群
	databasePtr database2.DataBaseItf
}

func MakeRedisHandler() *RedisServer {
	var databasePtr database2.DataBaseItf
	if len(config.PropPtr.Peers) > 0 {
		databasePtr = cluster.MakeCluster()
		logger.Info("集群数据库")
	} else {
		databasePtr = database.MakeDataBase()
		logger.Info("单机数据库")
	}
	return &RedisServer{
		// 数据存储:单机
		databasePtr: databasePtr,
	}
}

// Handle 客户端新连接处理
func (this *RedisServer) Handle(ctx context.Context, clientConn net.Conn) {

	// 整个服务关闭了不接收新的连接
	if this.closing.Get() {
		_ = this.CloseAll()
		return
	}
	this.activeConnMap.Store(clientConn, struct{}{})

	// 简单的回复
	//this.echoHandler(clientConn)

	// redis命令解析执行
	payloadPtrChn := protocol.ParseProtocol(clientConn)

	clientRedisConnPtr := redis.MakeRedisConn(clientConn)

	// 从通道中获取解执行后的命令回复--回发给客户端
	for payloadPtr := range payloadPtrChn {
		if payloadPtr.Err != nil {
			// 客户端连接断开
			if payloadPtr.Err == io.EOF {
				this.closeClient(clientConn)
				return
			}
			errReplyPtr := protocol.MakeErrReply(payloadPtr.Err.Error())
			_, err := clientConn.Write(errReplyPtr.ToBytes())
			// 写入失败关闭连接
			if err != nil {
				this.closeClient(clientConn)
				return
			}
			continue
		}
		if payloadPtr.DataReply == nil {
			logger.Error("empty payload")
			continue
		}
		protocolReplyPtr, ok := payloadPtr.DataReply.(*protocol.MultiBulkReply)
		if !ok {
			logger.Error("require multi bulk protocol")
			continue
		}
		// 在DB中执行指令
		reply := this.databasePtr.Exec(clientRedisConnPtr, protocolReplyPtr.ArgBytesSlc)
		_, _ = clientConn.Write(reply.ToBytes())
	}
}

// ECHO回复
func (this *RedisServer) echoHandler(clientConn net.Conn) {
	readerPtr := bufio.NewReader(clientConn)
	go func() {
		for {
			cmdBytes, err := readerPtr.ReadBytes('\n')
			if err != nil {
				if err == io.EOF {
					this.closeClient(clientConn)
					return
				} else {
					logger.Error(err.Error())
					return
				}
			}
			_, _ = clientConn.Write([]byte(cmdBytes))
		}
	}()
}

// CloseAll Close 关闭所有的客户端连接
func (this *RedisServer) CloseAll() error {
	logger.Info(constant.SERVER_LOG_TIP + "Shutdown! ")
	this.closing.Set(true)
	this.activeConnMap.Range(func(key, value interface{}) bool {
		clientConn, ok := key.(net.Conn)
		if !ok {
			logger.Info(fmt.Sprint(key) + "转换失败")
		}
		_ = clientConn.Close()
		logger.Info(constant.SERVER_LOG_TIP + "Shutdown : " + clientConn.RemoteAddr().String() + "连接关闭!")
		return true
	})
	this.databasePtr.Close()
	return nil
}

// 释放客户端连接
func (this *RedisServer) closeClient(clientConn net.Conn) {
	logger.Info(clientConn.RemoteAddr().String() + "连接关闭!")
	this.activeConnMap.Delete(clientConn)
}
